package org.example.consumer;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Collections;
import java.util.Properties;

/**
 * 自动提交 Automatic Commit
 * 执行：/opt/install/kafka/kafka/bin/kafka-console-producer.sh --broker-list kafka-123:9092,kafka-124:9092,kafka-125:9092 --topic test-topic1
 * 测试： kafka自动提交偏移量
 * 测试kafkaConsumer 消费偏移量
 * 备注：enable.auto.commit设置成true（默认为true），那么每过5s，消费者自动把从poll()方法接收到的最大的偏移量提交。提交的时间间隔由auto.commit.interval.ms控制，默认是5s
 * 自动提交的优点是方便，
 * 不足：但是可能会重复处理消息
 * @author youshang
 * @date 2021/01/26 18:38
 **/
public class LocalKafkaConsumerCommitTest1 {

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-123:9092,kafka-124:9092,kafka-125:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "group1");
        //--------------------------偏移量自动提交配置----begin---------------------------
        //是否开启消费组自动提交偏移量 默认为true
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        //和商品开启的配置配套使用，自动提交的时间间隔是多久，默认是 5000ms ，当前设置50s 在50秒之内关闭客户端可以清晰可见没次都在重复消费
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "50000");
        //----------------------------偏移量自动提交配置--end---------------------------

        // 当前配置如果生效的情况下有可能会出发rebalance(分区重平衡)
        //指定消费者被认定死亡之前可以与服务器断开连接的时间，默认是3s
        //props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");

        //从最小的偏移量开始消费
        //props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        //kafka默认使用String字符串序列化，如果使用了特殊的序列化需要进行特殊的反序列化
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        // 1.创建消费者
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(props);

        //2.订阅Topic
        //创建一个只包含单个元素的列表，Topic的名字叫作customerCountries
        kafkaConsumer.subscribe(Collections.singleton("test-topic1"));
        //支持正则表达式，订阅所有与test相关的Topic
        //consumer.subscribe("test.*");

        try {
            //3.轮询
            //消息轮询是消费者的核心API，通过一个简单的轮询向服务器请求数据，一旦消费者订阅了Topic，轮询就会处理所欲的细节，包括群组协调、partition再均衡、发送心跳
            //以及获取数据，开发者只要处理从partition返回的数据即可。
            //消费者是一个长期运行的程序，通过持续轮询向Kafka请求数据。在其他线程中调用consumer.wakeup()可以退出循环
            while (true) {
                //在100ms内等待Kafka的broker返回数据.超时参数指定poll在多久之后可以返回，不管有没有可用的数据都要返回
                ConsumerRecords<String, String> records = kafkaConsumer.poll(100);
                                          for (ConsumerRecord<String, String> record : records) {
                    //获取当前消息的偏移量
                    long offset = record.offset();
                    String key = record.key();
                    System.out.println("offset:"+ offset + "   key :" + key + "   value: " + record.value());
                }
            }
        }finally {
            //退出应用程序前使用close方法关闭消费者，网络连接和socket也会随之关闭，并立即触发一次再均衡
            kafkaConsumer.close();
        }
    }

}
